package spring.cloud.tasks.executor.name_space.executor.task;

import com.alibaba.fastjson.JSON;
import com.google.common.base.Strings;
import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
import org.springframework.util.PropertyPlaceholderHelper;
import spring.cloud.tasks.common.Threads;
import spring.cloud.tasks.executor.name_space.executor.task.task_item.*;
import spring.cloud.tasks.executor.name_space.executor.task.task_listener.task_configuration.TaskConfigurationService;
import spring.cloud.tasks.executor.name_space.executor.tasks_runtime_info.task_item_execute.*;
import spring.cloud.tasks.executor.utils.LogUtils;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.concurrent.ConcurrentHashMap;

@Slf4j
@Setter
@Getter
public class TaskExecuteEngine {
    public volatile boolean needStop = false;//外部尝试关闭任务执行器
    //
    protected String nameSpace;
    protected String executorId;
    protected String taskId;
    protected TaskExecutor taskExecutor;
    protected TaskConfigurationService taskConfigurationService;
    protected TaskItemExecuteInfoService taskItemExecuteInfoService;
    //
    private volatile boolean stopped = false;
    private volatile boolean forceStopped = false;
    private volatile boolean running = false;


    public final void startTaskItemExecuteThread() {
        if (isNeedStop()) {
            LogUtils.debug(log, taskId, "need stop to execute job [{}]", taskId);
            return;
        }
        LogUtils.debug(log, taskId, "Saturn start to execute job [{}]", taskId);
        try {
            running = true;
            TaskExecuteContext taskExecuteContext = getTaskExecuteContext();
            if (taskExecuteContext.getTaskItemExecuteRuntimeContextList().isEmpty()) {
                LogUtils.debug(log, taskId, "{} 's items of the executor is empty, do nothing about business.");
                return;
            }
            LogUtils.debug(log, taskId, " start to execute job [{}], sharding context:{}.", taskId, taskExecuteContext);
            startTaskItemExecuteThread(taskExecuteContext);
            LogUtils.debug(log, taskId, " finish to execute job [{}], sharding context:{}.", taskId, taskExecuteContext);
        } catch (Exception e) {
            LogUtils.warn(log, taskId, e.getMessage(), e);
        } finally {
            running = false;
        }
    }

    /**
     * 获取当前任务服务器运行时分片上下文.计算出当前能够运行的任务片
     *
     * @return 当前任务服务器运行时分片上下文
     */
    public TaskExecuteContext getTaskExecuteContext() {
        TaskExecuteContext taskExecuteContext = new TaskExecuteContext();
        taskExecuteContext.setNameSpace(nameSpace);
        taskExecuteContext.setExecutorId(executorId);
        taskExecuteContext.setTaskId(taskId);

        taskExecuteContext.setTaskConfiguration(taskConfigurationService.taskConfiguration);
        taskExecuteContext.setTaskItemTotalCount(taskConfigurationService.getTaskItemTotalCount());
        /**
         * 允许被分配的任务分片
         */
        ConcurrentHashMap<String, List<Integer>> taskInstanceIdToItemIndexListMap = taskExecutor.getTaskInstanceIdToItemIndexListMap();
        List<TaskItemExecuteRuntimeContext> list = getNeedRunningTaskItemExecuteRuntimeContextList(taskExecuteContext, taskInstanceIdToItemIndexListMap);
        taskExecuteContext.setTaskItemExecuteRuntimeContextList(list);
        taskExecuteContext.setPropertiesString(taskConfigurationService.getPropertiesString());
        //
        if (taskExecuteContext.getTaskInstanceIdToItemIndexListMap().isEmpty()) {
            return taskExecuteContext;
        }
        Map<Integer, String> taskItemIndexToTaskItemParameterMap = taskConfigurationService.getTaskItemIndexToTaskItemParameterMap();
        for (String taskInstanceId : taskInstanceIdToItemIndexListMap.keySet()) {
            for (Integer taskItemIndex : taskInstanceIdToItemIndexListMap.get(taskInstanceId)) {
                if (taskItemIndexToTaskItemParameterMap.containsKey(taskItemIndex)) {
                    taskExecuteContext.getTaskItemIndexToTaskItemParameterMap().put(taskItemIndex, taskItemIndexToTaskItemParameterMap.get(taskItemIndex));
                }
            }
        }
        taskExecuteContext.setSecondAfterTaskStopCommandInstructed(taskConfigurationService.taskConfiguration.getSecondAfterTaskStopCommandInstructed());
        if (taskConfigurationService.getTimeoutMinutes() > 0) {
            taskExecuteContext.setTimeoutMinutes(taskConfigurationService.getTimeoutMinutes());
        }
        return taskExecuteContext;
    }

    /**
     * @param taskInstanceIdToItemIndexListMap
     */
    private List<TaskItemExecuteRuntimeContext> getNeedRunningTaskItemExecuteRuntimeContextList(TaskExecuteContext taskExecuteContext, ConcurrentHashMap<String, List<Integer>> taskInstanceIdToItemIndexListMap) {
        List<TaskItemExecuteRuntimeContext> list = new ArrayList<TaskItemExecuteRuntimeContext>();
        TaskExecuteRuntimeContext taskExecuteRuntimeContext = TasksExecuteRuntimeContext.getTaskExecuteRuntimeContext(taskId);
        for (String taskInstanceId : taskInstanceIdToItemIndexListMap.keySet()) {
            TaskInstanceExecuteRuntimeContext TaskItemExecuteRuntimeContexts = taskExecuteRuntimeContext.getTaskInstanceExecuteRuntimeContext(taskInstanceId);
            for (Integer taskItemIndex : taskInstanceIdToItemIndexListMap.get(taskInstanceId)) {
                spring.cloud.tasks.executor.name_space.executor.tasks_runtime_info.task_item_execute.TaskItemExecuteRuntimeContexts taskItemExecuteRuntimeContexts = TaskItemExecuteRuntimeContexts.getTaskItemExecuteRuntimeContexts(taskItemIndex);
                if (taskItemExecuteRuntimeContexts.currentTaskItemExecuteRuntimeContext == null) {
                    //这个地方是创建任务的入口
                    TaskItemExecuteRuntimeContext taskItemExecuteRuntimeContext = bulid(taskExecuteContext, taskInstanceId, taskItemIndex, taskItemExecuteRuntimeContexts);
                    taskItemExecuteRuntimeContexts.currentTaskItemExecuteRuntimeContext = taskItemExecuteRuntimeContext;
                    list.add(taskItemExecuteRuntimeContext);
                    continue;
                } else {
                    TaskItemExecuteRuntimeContext taskItemExecuteRuntimeContext = taskItemExecuteRuntimeContexts.currentTaskItemExecuteRuntimeContext;
                    //检查当前任务分片是否完成
                    if (taskItemExecuteRuntimeContext.isFinished()) {
                        //
                        taskItemExecuteRuntimeContext = bulid(taskExecuteContext, taskInstanceId, taskItemIndex, taskItemExecuteRuntimeContexts);
                        taskItemExecuteRuntimeContexts.currentTaskItemExecuteRuntimeContext = taskItemExecuteRuntimeContext;
                        continue;
                    } else {
                        continue;
                    }

                }
            }
        }
        return list;
    }

    private TaskItemExecuteRuntimeContext bulid(TaskExecuteContext taskExecuteContext, String taskInstanceId, Integer taskItemIndex, TaskItemExecuteRuntimeContexts taskItemExecuteRuntimeContexts) {
        TaskItemExecuteRuntimeContext taskItemExecuteRuntimeContext = new TaskItemExecuteRuntimeContext(//
                taskExecuteContext.getNameSpace(), taskExecuteContext.getExecutorId(), taskExecuteContext.getTaskId(), taskInstanceId, taskItemIndex, taskInstanceId,//
                taskExecuteContext.getInitRenewalTimeMinutes(), taskExecuteContext.getMaxRenewalTimeMinutes());
        return taskItemExecuteRuntimeContext;
    }


    protected final void startTaskItemExecuteThread(final TaskExecuteContext taskExecuteContext) {
        // shardingItemParameters为参数表解析出来的Key/Value值
        Map<Integer, String> taskItemIndexToTaskItemParameterMap = taskExecuteContext.getTaskItemIndexToTaskItemParameterMap();
        LogUtils.info(log, taskId, "Job {} handle items: {}", taskId, taskItemIndexToTaskItemParameterMap);
        // items为需要处理的任务分片

        List<TaskItemExecuteRuntimeContext> list = taskExecuteContext.getTaskItemExecuteRuntimeContextList();
        LogUtils.info(log, taskId, "Job {} handle items: {}", taskId, JSON.toJSONString(list));
        for (TaskItemExecuteRuntimeContext taskItemExecuteRuntimeContext : list) {
            taskItemExecuteInfoService.registerTaskItemExecuteRuntimeContext(taskItemExecuteRuntimeContext);
        }


        //
        final String taskId = taskExecuteContext.getTaskId();
        final Integer timeoutMinutes = taskExecuteContext.getTimeoutMinutes();

        List<TaskItemExecute> taskItemExecuteList = new ArrayList<TaskItemExecute>();
        List<TaskItemExecuteTask> taskItemExecuteTaskList = new ArrayList<TaskItemExecuteTask>();
        for (TaskItemExecuteRuntimeContext taskItemExecuteRuntimeContext : list) {
            String threadName = taskItemExecuteRuntimeContext.createThreadName();
            taskItemExecuteRuntimeContext.setThreadName(threadName);
            //
            final Integer taskItemIndex = taskItemExecuteRuntimeContext.getTaskItemIndex();
            String taskItemParameter = taskItemIndexToTaskItemParameterMap.get(taskItemIndex);
            // 处理自定义参数
            String propertiesString = taskExecuteContext.getPropertiesString();
            final String $taskItemParameter = getReplaceTaskItemParameter(taskItemParameter, propertiesString); // 任务分片的对应值
            final Integer secondAfterTaskStopCommandInstructed = taskExecuteContext.getSecondAfterTaskStopCommandInstructed();


            TaskItemExecute taskItemExecute = new TaskItemExecute(taskExecutor, taskId, getTaskExecuteContext().getTaskConfiguration().getLinkId(), taskItemIndex, $taskItemParameter, secondAfterTaskStopCommandInstructed);
            taskItemExecute.bind(taskItemExecuteRuntimeContext);
            taskItemExecuteList.add(taskItemExecute);
            //
            TaskItemExecuteTask taskItemExecuteTask = new TaskItemExecuteTask(taskItemExecute);
            taskItemExecuteTask.setThreadName(threadName);
            taskItemExecuteTaskList.add(taskItemExecuteTask);

        }
        for (TaskItemExecuteTask taskItemExecuteTask : taskItemExecuteTaskList) {
            String threadName = taskItemExecuteTask.getThreadName();
            TaskItemExecute taskItemExecute = taskItemExecuteTask.getTaskItemExecute();

            taskItemExecute.setThreadName(threadName);
            TaskItemExecuteThread taskItemExecuteThread = new TaskItemExecuteThread(taskItemExecuteTask);
            taskItemExecuteThread.setDaemon(true);
            taskItemExecuteThread.setName(threadName);
            taskItemExecuteThread.start();
            //add id
            if (timeoutMinutes != null) {
                TaskItemExecuteTimeoutTaskSchedulerExecutor.schedule(taskExecuteContext.getExecutorId(), threadName, timeoutMinutes * 60 * 1000);
            }

        }
        //在整个执行的过程中 存在外部的线程主动和外面的ZK进行交互数据，检测状态。


    }

    /**
     * 获取替换后的任务分片执行值
     *
     * @param taskItemParameter 任务value
     * @param propertiesString  任务参数
     * @return 替换后的值
     */
    protected String getReplaceTaskItemParameter(String taskItemParameter, String propertiesString) {
        // 处理自定义参数
        /**
         * key1=value1
         * key2=value2
         * key3=${value4}/${value5}
         */
        Properties properties = parseProperties(propertiesString);
        int size = properties != null ? properties.size() : 0;
        final String string; // 任务分片的对应值
        if (size > 0) {
            // 有自定义参数, 解析完替换
            PropertyPlaceholderHelper propertyPlaceholderHelper = new PropertyPlaceholderHelper("{", "}");
            string = propertyPlaceholderHelper.replacePlaceholders(taskItemParameter, properties);
        } else {
            string = taskItemParameter;
        }
        return string;
    }

    public Properties parseProperties(String string) {
        if (Strings.isNullOrEmpty(string)) {
            return null;
        }
        Properties properties = new Properties();
        String[] strings = string.split(",");
        if (strings != null && strings.length > 0) {
            for (String $string : strings) {
                String[] splits = $string.split("=");
                if (splits != null && splits.length == 2) {
                    properties.put(splits[0].trim(), splits[1].trim());
                }
            }
        }


        return properties;
    }


    public void stop() {
        stopped = true;
    }


    public void forceStop() {
        forceStopped = true;
        TaskExecuteRuntimeContext taskExecuteRuntimeContext = TasksExecuteRuntimeContext.getTaskExecuteRuntimeContext(taskId);
        for (String taskInstanceId : TaskExecuteRuntimeContext.taskInstanceIdToTaskInstanceExecuteRuntimeContextMap.keySet()) {
            TaskInstanceExecuteRuntimeContext TaskItemExecuteRuntimeContexts = taskExecuteRuntimeContext.getTaskInstanceExecuteRuntimeContext(taskInstanceId);
            for (Integer taskItemIndex : TaskInstanceExecuteRuntimeContext.taskItemIndexToTaskItemExecuteRuntimeContextsMap.keySet()) {
                TaskItemExecuteRuntimeContexts taskItemExecuteRuntimeContexts = TaskItemExecuteRuntimeContexts.getTaskItemExecuteRuntimeContexts(taskItemIndex);
                if (taskItemExecuteRuntimeContexts.currentTaskItemExecuteRuntimeContext == null) {
                    continue;
                } else {
                    TaskItemExecuteRuntimeContext taskItemExecuteRuntimeContext = taskItemExecuteRuntimeContexts.currentTaskItemExecuteRuntimeContext;
                    taskItemExecuteRuntimeContext.needStop = true;//
                    //检查当前任务分片是否完成
                    if (!taskItemExecuteRuntimeContext.isFinished()) {
                        Thread thread = Threads.getThreadByhreadName(taskItemExecuteRuntimeContext.getThreadName());
                        if (thread == null) {
                            continue;
                        }
                        try {
                            TaskItemExecuteThread taskItemExecuteThread = (TaskItemExecuteThread) thread;
                            TaskItemExecuteTask taskItemExecuteTask = taskItemExecuteThread.getTaskItemExecuteTask();
                            if (taskItemExecuteTask == null) {
                                continue;
                            }
                            TaskItemExecute taskItemExecute = taskItemExecuteTask.getTaskItemExecute();
                            if (taskItemExecute == null) {
                                continue;
                            }
                            if (taskItemExecute.setForceStopWhenStatusInit()) {
                                LogUtils.info(log, taskId, "Force stop job, taskId:{}, item:{}", taskId, taskItemExecute.getTaskItemIndex());
                                new Thread(new Runnable() {
                                    @Override
                                    public void run() {
                                        TaskItemExecuteTaskKill.kill(taskItemExecuteTask);
                                    }
                                }).start();
                            }
                        } catch (Exception e) {
                            LogUtils.error(log, taskId, e.getMessage(), e);
                        }
                        continue;
                    } else {
                        continue;
                    }

                }
            }
        }

    }


}
